package com.spi.config;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.spi.mapper.UserMapper;
import domain.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

@Component("mqTransactionListener")
@Slf4j
public class MqTransactionListener implements TransactionListener {

    @Autowired
    private UserMapper userMapper ;

    /**
     * 半事务消息发送成功后回调此方法 执行本地事务
     * @param message
     * @param o
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        String transactionId = message.getTransactionId();
        log.info("事务id为：{}",transactionId);
        String mess = new String(message.getBody());
        log.info("开始执行本地事务：用户信息：" + mess);
        User user = JSONObject.parseObject(mess, User.class);
        LocalTransactionState state ; //用于告知mq二次确认结果
        try {
            //1:幂等性判断  必须进行幂等性判断
            //2:执行业务逻辑
            userMapper.insert(user);
            //3:将本地事务执行结果记录在事务记录表 消息id与事务id设为联合主键
            // 后续回调接口需用到该条数据 本次演示不加以考虑
            state = LocalTransactionState.COMMIT_MESSAGE; //提交
            //System.out.println(1/0);; //抛异常
        }catch (Exception e){
            log.info("本地事务执行异常：{}",e.getMessage());
            e.printStackTrace();
            state = LocalTransactionState.ROLLBACK_MESSAGE;//回滚
            throw new RuntimeException();
        }
        return state;
    }

    /**
     * 回查接口，unknow状态，业务方执行本地事务逻辑时间耗时过程或网络原因引起，会导致broker对事务消息进行回查
     * 默认回查此时15次，第一次回查间隔时间6s 后续每次间隔60s
     * @param messageExt
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String transactionId = messageExt.getTransactionId();
        log.info("事务id为：{}",transactionId);
        String mess = new String(messageExt.getBody());
        User user = JSONObject.parseObject(mess, User.class);
        log.info("调用本地事务回查接口：{}",mess);
        //通常这步应该去拿着消息的唯一id跟去消息事务记录表查询该条数据状态 成功or失败
        //若存在数据并且状态为成功则直接返回commit 若存在数据并且状态为失败则直接返回rollback
        //若无数据表示事务还未执行完毕 大事务较耗时 返回unknown 让mq过段时间再次回查

        //本次演示不加以考虑  只判断用户表是否有该条数据
        LambdaQueryWrapper<User> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        lambdaQueryWrapper.eq(User::getName,user.getName());
        List<User> users = userMapper.selectList(lambdaQueryWrapper);
        if (users.size() < 1){
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
